package com.xrui.hbase.filter;

import com.google.common.base.Objects;

import com.xrui.hbase.ColumnName;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.KeyOnlyFilter;

import java.io.IOException;

/**
 * <p>
 * Filter that replaces the value of each cell by its size, in number of bytes.
 * </p>
 * <p> Each cell size is encoded as a 32 bits integer (see
 * {@link org.apache.hadoop.hbase.util.Bytes#toInt(byte[])}).
 * </p>
 * <p>
 * In order to use this filter, you must override the reader schema of the affected columns
 * as a final fixed byte array of size 4.
 * </p>
 * <pre><tt>{@code
 *   final Table table = ...;
 *   final TableLayout layout = table.getLayout();
 *   final ColumnName column = ColumnName.create("family", "qualifier");
 *   final Map<ColumnName, CellSpec> overrides =
 *       ImmutableMap.<ColumnName, CellSpec>builder()
 *       .put(column, layout.getCellSpec(column)
 *           .setCellSchema(CellSchema.newBuilder()
 *               .setType(SchemaType.INLINE)
 *               .setStorage(SchemaStorage.FINAL)
 *               .setValue("{\"type\": \"fixed\", \"size\": 4, \"name\": \"Int32\"}")
 *               .build()))
 *       .build();
 *
 *   final TableReader reader = table.getReaderFactory().openTableReader(overrides);
 *   try {
 *     final DataRequest dataRequest = DataRequest.builder()
 *         .addColumns(ColumnsDef.create().withFilter(new CellByteSizeAsValueFilter()).add(column))
 *         .build();
 *     final EntityId eid = table.getEntityId(...);
 *     final RowData row = reader.create(eid, dataRequest);
 *     final GenericData.Fixed fixed32 = row.getMostRecentValue("family", "qualifier");
 *     final int cellSize = Bytes.toInt(fixed32.bytes());
 *     ...
 *   } finally {
 *     reader.close();
 *   }
 *   }</tt></pre>
 * <p>
 * {@link org.apache.hadoop.hbase.filter.FilterList}.
 * If you need to resolve this filter to specific columns, you must currently send separate create
 * requests.
 * </p>
 */
public final class CellByteSizeAsValueFilter extends ColumnFilter {
    private static final long serialVersionUID = 1L;

    /**
     * All StripValueColumnFilter instances are the same: generate hash-code ahead of time.
     */
    private static final int HASH_CODE =
        new HashCodeBuilder().append(CellByteSizeAsValueFilter.class).toHashCode();

    /**
     * {@inheritDoc}
     */
    @Override
    public Filter toHBaseFilter(ColumnName columnName/*, Context context*/) throws IOException {
        return new KeyOnlyFilter(true);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object other) {
        // All StripValueRowFilters are the same.
        return other instanceof CellByteSizeAsValueFilter;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return HASH_CODE;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return Objects.toStringHelper(CellByteSizeAsValueFilter.class).toString();
    }
}
